package com.asap.demo.sink;

import com.asap.demo.function.sink.DealElasticsearchSinkFunction;
import com.asap.demo.utils.es.RetryRequestFailureHandler;
import com.asap.rule.StandardEvent;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;

import java.util.List;

/**
 * @author wangbh
 * @Description: TODO
 * @date 2021/11/26 17:17
 */
public class DealElasticsearchSink {
    List<HttpHost> esAddresses;
    int bulkSize;

    public ElasticsearchSink<StandardEvent> build(List<HttpHost> esAddresses, int bulkSize) {
        this.esAddresses = esAddresses;
        this.bulkSize = bulkSize;
        return  getJsonObjectEsSinkBuilder();
    }

    private ElasticsearchSink<StandardEvent> getJsonObjectEsSinkBuilder() {
        ElasticsearchSink.Builder<StandardEvent> esSinkBuilder = new ElasticsearchSink.Builder<>(this.esAddresses, new DealElasticsearchSinkFunction());
        esSinkBuilder.setBulkFlushMaxActions(this.bulkSize);
        esSinkBuilder.setFailureHandler(new RetryRequestFailureHandler());
        return esSinkBuilder.build();
    }
}
